package xiaojian.toolkit.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.util.AttributeKey;

import org.apache.commons.pool2.ObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import xiaojian.toolkit.bean.tlv.annotation.HttpTLVAttribute;
import xiaojian.toolkit.netty.handler.codec.http.response.DefaultIdentifiableHttpResponse;
import xiaojian.toolkit.netty.transport.PoolableSender;
import xiaojian.toolkit.netty.util.NettyConstants;
import xiaojian.toolkit.util.Identifyable;
import xiaojian.toolkit.util.MutableIdentifyable;

import java.util.UUID;

public class HttpClient implements PoolableSender {
	private static Logger logger = LoggerFactory.getLogger(HttpClient.class);

	private static final NioEventLoopGroup NIO_EVENT_LOOP_GROUP = new NioEventLoopGroup();

	private static final AttributeKey<UUID> CTX_UUID = AttributeKey
			.valueOf("CTX_UUID");

	private ChannelInitializer httpClientInitializer;
	private Bootstrap bootstrap = new Bootstrap();
	private String remoteAddress;
	private int remotePort;
	private String localAddress = "0.0.0.0";
	private int localPort;
	private int connectTimeoutMillis = 10000; // 10 seconds

	private boolean connected;
	private Channel channel;
	private boolean isReturnedToPool;

	private ObjectPool pool;
	private HttpClient client = this;

	private final class ChannelObserver extends ChannelInboundHandlerAdapter {

		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg)
				throws Exception {
			returnClientToPool("messageReceived");
			UUID uuid = ctx.channel().attr(CTX_UUID).get();
			if (uuid != null) {
				ctx.channel().attr(CTX_UUID).remove();
				if (msg instanceof FullHttpResponse) {
					DefaultIdentifiableHttpResponse response = new DefaultIdentifiableHttpResponse();
					response.setUuid(uuid);
					response.setHttpResponse((FullHttpResponse) msg);
					super.channelRead(ctx, response);
					return;
				} else if (msg instanceof MutableIdentifyable) {
					((MutableIdentifyable) msg).setIdentification(uuid);
				}
			}

			super.channelRead(ctx, msg);
		}

		@Override
		public void channelInactive(ChannelHandlerContext ctx) throws Exception {
			connected = false;
			channel = null;
			returnClientToPool("channelInactive");
			super.channelInactive(ctx);
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
				throws Exception {
			logger.error("exceptionCaught ", cause);
			returnClientToPool("exceptionCaught");
			super.exceptionCaught(ctx, cause);
		}

		@SuppressWarnings("unchecked")
		private void returnClientToPool(String reason) throws Exception {
			if (pool == null) {
				logger.info("http client pool is not set,there will be pool leak if you are using a http client pool.");
			} else if (client.isReturnedToPool()) {
				logger.debug(
						"{},client {} is returned to pool already, so don't return it again",
						reason, client);
			} else {
				client.setReturnedToPool(true);
				pool.returnObject(client);
				logger.debug("{},and returned {} to pool", reason, client);
			}
		}
	}

	public void send(Object message) {
		if (message == null)
			return;
		if (message instanceof Identifyable) { // 在Channel中加上uuid
			channel.attr(CTX_UUID).set(
					((Identifyable) message).getIdentification());
		}
		HttpTLVAttribute attr = message.getClass().getAnnotation(
				HttpTLVAttribute.class);
		if (attr != null) {
			channel.attr(NettyConstants.CTX_HTTP_RESP_CLASS).set(
					attr.responseType());
		}
		channel.writeAndFlush(message);
	}

	public HttpClient connect() throws Exception {
		bootstrap
				.group(NIO_EVENT_LOOP_GROUP)
				.channel(NioSocketChannel.class)
				.localAddress(localAddress, localPort)
				.remoteAddress(remoteAddress, remotePort)
				.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
						connectTimeoutMillis).handler(httpClientInitializer);

		ChannelFuture future = bootstrap.connect();
		logger.debug("connection opened,waiting for channel connected!");
		channel = future.await().channel();
		if (channel.pipeline().get("handler") != null) {// 确保在将http应答提交给业务层之前将uuid装配好
			channel.pipeline().addBefore("handler", "observer",
					new ChannelObserver());
		} else {
			channel.pipeline().addLast("observer", new ChannelObserver());
		}

		logger.debug("new http client channel established!");

		return this;
	}

	public void close() {
		if (channel != null) {
			channel.close();
		}
	}

	@Override
	public String toString() {
		return "httpClient[" + channel + "]";
	}

	public String getRemoteAddress() {
		return remoteAddress;
	}

	public void setRemoteAddress(String remoteAddress) {
		this.remoteAddress = remoteAddress;
	}

	public int getRemotePort() {
		return remotePort;
	}

	public void setRemotePort(int remotePort) {
		this.remotePort = remotePort;
	}

	public boolean isConnected() {
		return connected;
	}

	public Channel getChannel() {
		return channel;
	}

	public String getLocalAddress() {
		return localAddress;
	}

	public void setLocalAddress(String localAddress) {
		this.localAddress = localAddress;
	}

	public int getLocalPort() {
		return localPort;
	}

	public void setLocalPort(int localPort) {
		this.localPort = localPort;
	}

	public void setPool(ObjectPool pool) {
		this.pool = pool;
	}

	public void setHttpClientInitializer(
			ChannelInitializer httpClientInitializer) {
		this.httpClientInitializer = httpClientInitializer;
	}

	public boolean isReturnedToPool() {
		return isReturnedToPool;
	}

	public void setReturnedToPool(boolean returnedToPool) {
		isReturnedToPool = returnedToPool;
	}

	public void setConnectTimeoutMillis(int connectTimeoutMillis) {
		this.connectTimeoutMillis = connectTimeoutMillis;
	}
}
